﻿using System;
using System.Collections.Generic;

using Apache.Cassandra;
using Client = Apache.Cassandra.Cassandra.Client;

using RexToy.Cassandra.Meta;

namespace RexToy.Cassandra
{
    class CounterColumnFamily<K, C> : ColumnFamilyBase<K, C>, ICounterColumnFamily<K, C>
    {
        public CounterColumnFamily(string keyspace, string columnFamily, ColumnFamilyInfo info, Client c, ConsistencyLevel read, ConsistencyLevel write)
            : base(keyspace, columnFamily, info, c, read, write)
        {
        }

        public void Incr(K key, C column, long value)
        {
            var parent = Utils.CreateColumnParent(this.Name);
            CounterColumn c = new CounterColumn();
            c.Name = this.GetColumnBytes(column);
            c.Value = value;
            _c.add(this.GetKeyBytes(key), parent, c, _write);
        }

        public void Remove(K key)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, null);
            _c.remove_counter(this.GetKeyBytes(key), cp, _write);
        }

        public void Remove(K key, C column)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(column));
            _c.remove_counter(this.GetKeyBytes(key), cp, _write);
        }

        public long Select(K key, C column)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(column));
            var c = _c.get(this.GetKeyBytes(key), cp, _read);
            return c.Counter_column.Value;
        }

        public ICounterRow<K, C> Select(K key, int count = 10, bool reversed = false)
        {
            return this._Select(key, default(C), default(C), count, reversed);
        }

        public ICounterRow<K, C> Select(K key, C start, C finish, int count = 10, bool reversed = false)
        {
            return this._Select(key, start, finish, count, reversed);
        }

        private ICounterRow<K, C> _Select(K key, C start, C finish, int count, bool reversed)
        {
            SlicePredicate predicate = Utils.CreateSlicePredicate(this.GetColumnBytes(start), this.GetColumnBytes(finish), count, reversed);
            ColumnParent cp = Utils.CreateColumnParent(this.Name);

            var list = _c.get_slice(this.GetKeyBytes(key), cp, predicate, _read);

            var collection = this.CreateColumns(list);
            return this.CreateRow(key, collection);
        }

        protected ICounterRow<K, C> CreateRow(K key, ICounterColumnCollection<C> collection)
        {
            return new CounterRow<K, C>(key, collection);
        }

        private ICounterColumnCollection<C> CreateColumns(List<ColumnOrSuperColumn> list)
        {
            var collection = new CounterColumnCollection<C>();
            foreach (var col in list)
            {
                collection.AddColumn(this.CreateColumn(col.Counter_column));
            }
            return collection;
        }

        private ICounterColumn<C> CreateColumn(CounterColumn column)
        {
            var col = this.GetColumnData(column.Name);
            return new CounterColumn<C>(col, column.Value);
        }
    }
}
